package se.kth.ict.id2203.pb;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.flp2p.FairLossPointToPointLink;
import se.kth.ict.id2203.flp2p.Flp2pDeliver;
import se.kth.ict.id2203.flp2p.Flp2pSend;
import se.kth.ict.id2203.sub.UnBroadcast;
import se.kth.ict.id2203.sub.UnDeliver;
import se.kth.ict.id2203.sub.UnreliableBroadcast;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.address.Address;
import se.sics.kompics.launch.Topology;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;

public class LPB extends ComponentDefinition {
	
	Negative<ProbabilisticBroadcast> pb = provides(ProbabilisticBroadcast.class);
	Positive<UnreliableBroadcast> sub = requires(UnreliableBroadcast.class);	
	Positive<FairLossPointToPointLink> flp2p = requires(FairLossPointToPointLink.class);	
	Positive<Timer> timer = requires(Timer.class);
	
	private Address self;
	private Topology topology;
	private Set<Address> processSet;
	private Set<Address> neighborSet;
	private Map<Address, Long> next;
	private long lsn;
	
//	private Set<UnDeliver> pending;
//	private Set<UnDeliver> stored;
	
	private Set<Data> pending;
	private Set<Data> stored;
	
	
	private int fanout;
	private double store_threshold;
	private int maxrounds;
	private long delta;
	
	
	private static final Logger logger =
			LoggerFactory.getLogger(LPB.class);
	
	public LPB() {
		logger.info("LazyPB created.");
		subscribe(handleInit, control);
		subscribe(handlePbBroadcast, pb);
		subscribe(handleUnDeliver, sub);
		subscribe(handleRequest, flp2p);
		subscribe(handleDataMessage, flp2p);
		subscribe(handleTimeout, timer);
		
	}
	
	Handler<LPBInit> handleInit = new Handler<LPBInit>() {
		public void handle(LPBInit event) {
			logger.info("LazyPB started.");
			topology = event.getTopology();
			self = topology.getSelfAddress();
			processSet = topology.getAllAddresses();
			neighborSet = topology.getNeighbors(self);
			
			fanout = event.getFanout();
			store_threshold = event.getStore_threshold();
			maxrounds = event.getMaxrounds();
			delta = event.getDelta();
			
			next = new HashMap<Address, Long>();
			for (Address process : processSet) {
				next.put(process, 1L);
			}			
			lsn = 0;
			
//			pending = new HashSet<UnDeliver>();
//			stored = new HashSet<UnDeliver>();
			
			pending = new HashSet<Data>();
			stored = new HashSet<Data>();
			
		}
	};
	
	Handler<PbBroadcast> handlePbBroadcast = new Handler<PbBroadcast>() {
		public void handle(PbBroadcast event) {
			String message = event.getMessage();
			lsn = lsn + 1;
			trigger(new UnBroadcast(self, message, lsn), sub);
			logger.info("Broadcasting message: " + message);
		}
	};
	
	Handler<UnDeliver> handleUnDeliver = new Handler<UnDeliver>() {
		public void handle(UnDeliver event) {
			
			long sn = event.getSn();
			Address src = event.getDataSource();
			String message = event.getMessage();
			Data data = new Data(src, message, sn);
						
			logger.info("UnDeliver event received from " + src + " SN:" + sn + " Message:" + message );
			// random([0, 1])
			Random r = new Random();
			Double rannum = r.nextInt(1001) / 1000.0; 
			
			if (rannum > store_threshold) {
//				stored.add(event);
				
				stored.add(data);
				
				logger.info("Message stored!");
			}
			
			if (sn == next.get(src)) {
				next.put(src, sn + 1);
				trigger(new PbDeliver(src, message), pb);
				
			} else if (sn > next.get(src)) {
//				pending.add(event);
				
				pending.add(data);
				
				for (long missing = next.get(src); missing <= (sn-1); missing++ ) {
					
					boolean exist = false;
					
//					for (UnDeliver pend : pending) {
					
					for (Data pend : pending) {
						long pendSn = pend.getSn();
						Address pendSrc = pend.getDataSource();
						
						if ((pendSn == missing) && (pendSrc.equals(src))) {
							exist = true;
							break;
						} else {
							exist = false;
						}
					}
					
					if (!exist) {
						gossip(new RequestMessage(self, self, src, missing, maxrounds-1));
						
					}
					
					
				}
				
				// startTimer
				ScheduleTimeout st = new ScheduleTimeout(delta);
				st.setTimeoutEvent(new LPBTimeout(st, src, sn));
				trigger(st, timer);	
				
			} // else if
		}
	};
	
	Handler<RequestMessage> handleRequest = new Handler<RequestMessage>() {
		public void handle (RequestMessage event) {
			
			Address s = event.getDataSource();
			Address q = event.getReqSender();
			Address p = event.getSource();
			long sn = event.getMissing();
			int r = event.getRounds();
			
			logger.info("RequestMessage received from " + p + " \n [REQUEST, Sender:"
			+ q + ", DataSource:" + s + ", SN:" + sn + ", Round:" + r + "]");
			
			boolean exist = false;
			String message = null;
			
//			for (UnDeliver store : stored) {
			
			for (Data store : stored) {
				long storeSn = store.getSn();
				Address storeSrc = store.getDataSource();
				
				if ((storeSn == sn) && (storeSrc.equals(s))) {
					exist = true;
					message = store.getMessage();
					logger.info("Message SN: " + sn + " from" + s + "is existed!");
					break;
				} else {
					exist = false;
				}
			}
			
			if (exist) {
				trigger(new Flp2pSend(q, new DataMessage(self, s, message, sn)), flp2p);
				logger.info("Sending DataMessage(" + s + ", SN:" + sn + " to " + q);
			} else if (r > 0) {
				gossip(new RequestMessage(self, q, s, sn, r-1));
			}
		}
	};
	
	Handler<DataMessage> handleDataMessage = new Handler<DataMessage>() {
		public void handle (DataMessage event) {
			
			Address s = event.getDataSource();
			Address p = event.getSource();
			String m = event.getMessage();
			long sn = event.getSn();
			
//			pending.add(new UnDeliver(s, m, sn));
			pending.add(new Data(s, m, sn));
			
			logger.info("DataMessage received from " + p + " ! DataSource: " + s + " SN: " + sn);
			
			boolean exist = true;
			long pendSn = 0;
			Address pendSrc = null;
//			UnDeliver temp = null;
			Data temp = null;
			String msg = null;
			
			while (exist) {
				
				if (pending.size() == 0) {
					break;
				}
				
//				for (UnDeliver pend : pending) {
				
				for (Data pend : pending) {
					pendSn = pend.getSn();
					pendSrc = pend.getDataSource();
					
					if ((pendSn == next.get(s)) && (pendSrc.equals(s))) {
						exist = true;
						temp = pend;
						msg = pend.getMessage();
						break;
					} else {
						exist = false;
					}
				}
				
				if (exist) {
					next.put(pendSrc, pendSn + 1);
					pending.remove(temp);
					trigger(new PbDeliver(s, msg), pb);
					
				}
			}
			
		}
	};
	
	Handler<LPBTimeout> handleTimeout = new Handler<LPBTimeout>() {
		public void handle(LPBTimeout event) {
			Address s = event.getS();
			long sn = event.getSn();
			
			if (sn > next.get(s)) {
				next.put(s, sn+1);
			}
			
			logger.info("LPBTimeout! Next[" + s + "]: " + next.get(s));
		}
	};
	
	
	public Set<Address> pickTargets(int k) {
		Set<Address> targets = new HashSet<Address>();
		Address candidate;
		while (targets.size() < k) {
			candidate = random(neighborSet);
			if (!targets.contains(candidate)) {
				targets.add(candidate);
			}
		}
		return targets;
	}
	
	public Address random(Set<Address> randomSet) {
		int size = randomSet.size();
		Random r = new Random();
		Object[] randomArray = randomSet.toArray();
		return (Address)randomArray[r.nextInt(size)];
	}
	
	public void gossip(Flp2pDeliver msg) {
		Set<Address> temp = pickTargets(fanout);
		for (Address t : temp) {
			trigger(new Flp2pSend(t, msg), flp2p);
			logger.info("Gossiping to " + t + "...");
		}
	}

}
